﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

using System.Threading;
using System.Collections.Concurrent;
using System.Data.SqlClient;
using System.Data;

namespace ExtensionClassLibrary
{
    /// <summary> 
    /// SqlServer数据访问帮助类 
    /// </summary> 
    public sealed class SqlHelper
    {
        /// <summary>
        /// 连接池 默认为100，可改 ;min pool size=5;max pool size=512;
        /// </summary>
        private static string _connStr = string.Format("User ID={2};Password={1};Initial Catalog=Trade;Data Source={0};", System.Configuration.ConfigurationManager.AppSettings["DBIP"].ToString(), System.Configuration.ConfigurationManager.AppSettings["DBPsss"].ToString(), System.Configuration.ConfigurationManager.AppSettings["DBAccount"].ToString());
        public static bool IsValid = false;
        /// <summary>
        /// SQL 持行超时,默认2000s,可配
        /// </summary>
        public static int _CommandTimeout = 2000;
        /// <summary>
        /// sql BulkCopy 持行超时,默认6000,可配
        /// </summary>
        public static int _BulkCopyTimeout = 6000;
        public static string IsUseAsynchronousDBFlag = System.Configuration.ConfigurationManager.AppSettings["IsUseAsynchronousDBFlag"].NullToStr();
        private static readonly object lockDBObject = new object();
        public static string sqlstrNopasswd;
        /// <summary>
        /// 异步调用限制
        /// </summary>
        public static List<string> toUseAsyncDBFlagList = new List<string>() {};
        #region 私有构造函数和方法

        public static Action<string> actionDBPushFlag;
        #region DBLogQueue
        private static bool _start = true;
        private static ConcurrentQueue<DBQueueData> _commandQueue = new ConcurrentQueue<DBQueueData>();
        #endregion

        private SqlHelper()
        {
            int tmpTimeout = 2000;
            int.TryParse(ConfigHelp.GetAppConfig("CommandTimeout"), out tmpTimeout);
            _CommandTimeout = tmpTimeout;

            int tmpBCTimeout = 6000;
            int.TryParse(ConfigHelp.GetAppConfig("BulkCopyTimeout"), out tmpTimeout);
            _BulkCopyTimeout = tmpBCTimeout;


        }
        public static string GetSqlStrNoPasswd()
        {
            if (!sqlstrNopasswd.IsNullOrWhiteSpace2())
            {
                return sqlstrNopasswd;
            }
            _connStr.SplitTrim(';').ForEach((a) =>
            {
                if (!a.StartsWith("Password"))
                {
                    sqlstrNopasswd += a + ";";
                }
            });
            return sqlstrNopasswd;
        }
        /// <summary> 
        /// 将SqlParameter参数数组(参数值)分配给SqlCommand命令. 
        /// 这个方法将给任何一个参数分配DBNull.Value; 
        /// 该操作将阻止默认值的使用. 
        /// </summary> 
        /// <param name="command">命令名</param> 
        /// <param name="commandParameters">SqlParameters数组</param> 
        private static void AttachParameters(SqlCommand command, SqlParameter[] commandParameters)
        {
            if (command == null) throw new ArgumentNullException("command");
            if (commandParameters != null)
            {
                foreach (SqlParameter p in commandParameters)
                {
                    if (p != null)
                    {
                        // 检查未分配值的输出参数,将其分配以DBNull.Value. 
                        if ((p.Direction == ParameterDirection.InputOutput || p.Direction == ParameterDirection.Input) &&
                            (p.Value == null))
                        {
                            p.Value = DBNull.Value;
                        }
                        command.Parameters.Add(p);
                    }
                }
            }
        }



        /// <summary> 
        /// 预处理用户提供的命令,数据库连接/事务/命令类型/参数 
        /// </summary> 
        /// <param name="command">要处理的SqlCommand</param> 
        /// <param name="connection">数据库连接</param> 
        /// <param name="transaction">一个有效的事务或者是null值</param> 
        /// <param name="commandType">命令类型 (存储过程,命令文本, 其它.)</param> 
        /// <param name="commandText">存储过程名或都T-SQL命令文本</param> 
        /// <param name="commandParameters">和命令相关联的SqlParameter参数数组,如果没有参数为'null'</param> 
        /// <param name="mustCloseConnection"><c>true</c> 如果连接是打开的,则为true,其它情况下为false.</param> 
        private static void PrepareCommand(SqlCommand command, SqlConnection connection, SqlTransaction transaction, CommandType commandType, string commandText, SqlParameter[] commandParameters, out bool mustCloseConnection)
        {
            if (command == null) throw new ArgumentNullException("command");
            if (commandText == null || commandText.Length == 0) throw new ArgumentNullException("commandText");

            // If the provided connection is not open, we will open it 
            if (connection.State != ConnectionState.Open)
            {
                mustCloseConnection = true;
                connection.Open();
            }
            else
            {
                mustCloseConnection = false;
            }

            // 给命令分配一个数据库连接. 
            command.Connection = connection;

            // 设置命令文本(存储过程名或SQL语句) 
            command.CommandText = commandText;

            // 分配事务 
            if (transaction != null)
            {
                if (transaction.Connection == null) throw new ArgumentException("The transaction was rollbacked or commited, please provide an open transaction.", "transaction");
                command.Transaction = transaction;
            }

            // 设置命令类型. 
            command.CommandType = commandType;

            // 分配命令参数 
            if (commandParameters != null)
            {
                AttachParameters(command, commandParameters);
            }
            return;
        }

        #endregion 私有构造函数和方法结束

        #region 数据库连接
        /// <summary> 
        /// 一个有效的数据库连接字符串 
        /// min pool size=5;max pool size=512;Async=true;Connect Timeout=360;
        /// </summary> 
        /// <returns></returns> 
        public static string GetConnSting()
        {
            if (IsUseAsynchronousDBFlag == "1")
            {
                return $"{_connStr}Async=true;min pool size=5;max pool size=1024;Connect Timeout=600;";
            }
            return _connStr;//ConfigurationManager.ConnectionStrings["ConStr"].ConnectionString;
        }
        /// <summary> 
        /// 一个有效的数据库连接对象 
        /// </summary> 
        /// <returns></returns> 
        public static SqlConnection GetConnection()
        {
            SqlConnection Connection = new SqlConnection(SqlHelper.GetConnSting());
            return Connection;
        }

        public static bool TestDB()
        {
            bool state = true;
            SqlConnection con = GetConnection();
            try
            {
                con.Open();
                if (con.State == System.Data.ConnectionState.Open)
                {
                    IsValid = true;
                    state = true;
                }
                StartThread();
            }
            catch (SqlException)
            {
                state = false;
                IsValid = false;
            }
            finally
            {
                con.Close();
            }
            return state;
        }
        #endregion

        #region ExecuteNonQuery命令
        /// <summary> 
        /// 执行指定数据库连接对象的命令 
        /// </summary> 
        /// <remarks> 
        /// 示例:  
        ///  int result = ExecuteNonQuery(conn, CommandType.StoredProcedure, "PublishOrders", new SqlParameter("@prodid", 24)); 
        /// </remarks> 
        /// <param name="connection">一个有效的数据库连接对象</param> 
        /// <param name="commandType">命令类型(存储过程,命令文本或其它.)</param> 
        /// <param name="commandText">T存储过程名称或T-SQL语句</param> 
        /// <param name="commandParameters">SqlParamter参数数组</param> 
        /// <returns>返回影响的行数</returns> 
        public static int ExecuteNonQuery(SqlConnection connection, DBCommand dc, CommandType commandType, string commandText, params SqlParameter[] commandParameters)
        {
            //Log.Info("调用存储过程：" + commandText);
            //if (commandParameters != null)
            //    foreach (SqlParameter item in commandParameters)
            //    {
            //        if (item.Direction == ParameterDirection.Input)
            //            Log.Info(item.ParameterName + "  " + item.Value.ToString());
            //    }

            try
            {
                if (IsUseAsynchronousDBFlag == "1" && toUseAsyncDBFlagList.Contains(commandText))
                {
                    ExecuteNonQueryAsync(connection, dc, commandType, commandText, commandParameters);
                    return 1;
                }
                _commandQueue.Enqueue(new DBQueueData() { type = 1, logName = String.Intern("DBLog"), sqlParameters = commandParameters, commandText = commandText });
                //记录db log
                ////LogHelper.WriteLog("DBLog", FormatSqlParam(commandText, commandParameters));

                if (connection == null) throw new ArgumentNullException("connection");

                // 创建SqlCommand命令,并进行预处理 
                SqlCommand cmd = new SqlCommand();
                cmd.CommandTimeout = _CommandTimeout;
                bool mustCloseConnection = false;
                PrepareCommand(cmd, connection, (SqlTransaction)null, commandType, commandText, commandParameters, out mustCloseConnection);

                // Finally, execute the command 
                int retval = cmd.ExecuteNonQuery();

                // 清除参数,以便再次使用. 
                cmd.Parameters.Clear();
                if (mustCloseConnection)
                    connection.Close();
                //db处理成功，写dblog
                if (dc != null && dc.Num > -1)
                {

                    ////LogHelper.WriteLog(dc.SuccessDBlogFilePath, $"|{dc.Num}|db");
                    actionDBPushFlag?.Invoke($"|{dc.Num}|db");

                }
                return retval;
            }
            catch (Exception ex)
            {
                //SystemEvent.GetInstance().OnSystemMessageEvent("数据库访问错误！ExecuteNonQuery" + commandText + "---" + ex.ToString(), true, 2);
                //LogHelper.Error(FormatSqlParam(commandText, commandParameters));
                return 0;
            }
        }
        public static int BeginExecuteNonQuery(SqlConnection connection, CommandType commandType, string commandText, params SqlParameter[] commandParameters)
        {
            try
            {
                _commandQueue.Enqueue(new DBQueueData() { type = 1, logName = String.Intern("DBLogAsyncA"), sqlParameters = commandParameters, commandText = commandText });
                //记录db log
                ////LogHelper.WriteLog("DBLogAsyncA", FormatSqlParam(commandText, commandParameters));

                if (connection == null) throw new ArgumentNullException("connection");

                // 创建SqlCommand命令,并进行预处理 
                SqlCommand cmd = new SqlCommand();
                cmd.CommandTimeout = _CommandTimeout;
                bool mustCloseConnection = false;
                PrepareCommand(cmd, connection, (SqlTransaction)null, commandType, commandText, commandParameters, out mustCloseConnection);

                AsyncCallback callback = new AsyncCallback(CallbackHandlerEndExecuteNonQuery);
                cmd.BeginExecuteNonQuery(callback, cmd);
                return 1;
            }
            catch (Exception ex)
            {
                connection.Close();
                connection.Dispose();
                //SystemEvent.GetInstance().OnSystemMessageEvent("数据库访问错误！BeginExecuteNonQuery" + commandText + "---" + ex.ToString(), true, 2);
                //LogHelper.Error(FormatSqlParam(commandText, commandParameters));
                return 0;
            }
        }
        public static void CallbackHandlerEndExecuteNonQuery(IAsyncResult result)
        {
            try
            {
                using (SqlCommand cmd = result.AsyncState as SqlCommand)
                {
                    try
                    {
                        var getResult = cmd.EndExecuteNonQuery(result);
                        if (getResult < 1)
                        {
                            writeErrorDBLog(cmd.Parameters[cmd.Parameters.Count - 1], cmd.CommandText);
                        }
                    }
                    catch (Exception ex)
                    {
                        var commandText = cmd.CommandText;
                        //SystemEvent.GetInstance().OnSystemMessageEvent("数据库访问错误！EndExecuteNonQuery" + commandText + "---" + ex.ToString(), true, 2);
                        //LogHelper.Error(FormatSqlParam(commandText, cmd.Parameters));
                    }
                    finally
                    {
                        cmd.Connection.Close();
                        cmd.Connection.Dispose();
                    }
                }

            }
            catch (Exception ex)
            {

                //LogHelper.Error(ex);
            }

        }
        public static async Task<int> ExecuteNonQueryAsync(SqlConnection connection, DBCommand dc, CommandType commandType, string commandText, params SqlParameter[] commandParameters)
        {
            //Log.Info("调用存储过程：" + commandText);
            //if (commandParameters != null)
            //    foreach (SqlParameter item in commandParameters)
            //    {
            //        if (item.Direction == ParameterDirection.Input)
            //            Log.Info(item.ParameterName + "  " + item.Value.ToString());
            //    }

            try
            {
                if (connection == null) throw new ArgumentNullException("connection");
                // 创建SqlCommand命令,并进行预处理  
                using (SqlCommand cmd = new SqlCommand())
                using (connection)
                {
                    //记录db log
                    ////LogHelper.WriteLog("DBLogAsyncB", FormatSqlParam(commandText, commandParameters));
                    _commandQueue.Enqueue(new DBQueueData() { type = 2, logName = String.Intern("DBLogAsyncB"), sqlParameters = commandParameters, commandText = commandText });

                    bool mustCloseConnection = false;
                    PrepareCommand(cmd, connection, (SqlTransaction)null, commandType, commandText, commandParameters, out mustCloseConnection);

                    cmd.CommandTimeout = _CommandTimeout;
                    // Finally, execute the command 
                    var retval = await cmd.ExecuteNonQueryAsync();

                    if (retval < 1)
                    {
                        writeErrorDBLog(cmd.Parameters[cmd.Parameters.Count - 1], cmd.CommandText);
                    }

                    //db处理成功，写dblog
                    if (dc != null && dc.Num > -1)
                    {
                        //LogHelper.WriteLog(dc.SuccessDBlogFilePath, $"|{dc.Num}|dbasync");
                        actionDBPushFlag?.Invoke($"|{dc.Num}|dbasync");

                    }

                    return retval;
                }
            }
            catch (Exception ex)
            {
                //SystemEvent.GetInstance().OnSystemMessageEvent("数据库访问错误！ExecuteNonQueryAsync" + commandText + "---" + ex.ToString(), true, 2);
                //LogHelper.Error(FormatSqlParam(commandText, commandParameters));
                return 0;
            }
        }
        //public static int ExecuteNonQuery(string sql, CommandType cmdType, params SqlParameter[] pms)
        //{
        //    try
        //    {
        //        return Microsoft.ApplicationBlocks.Data.SqlHelper.ExecuteNonQuery(_connStr, cmdType, sql, pms);
        //    }
        //    catch (Exception ex)
        //    {
        //        //SystemEvent.GetInstance().OnSystemMessageEvent("数据库访问错误！ExecuteNonQuery" + sql + "---" + ex.ToString());
        //        return 0;
        //    }
        //}

        /// <summary> 
        /// 执行指定数据库连接对象的命令 
        /// </summary> 
        /// <remarks> 
        /// 示例:  
        ///  int result = ExecuteNonQuery(conn, CommandType.StoredProcedure, "PublishOrders", new SqlParameter("@prodid", 24)); 
        /// </remarks> 
        /// <param name="connection">一个有效的数据库连接对象</param> 
        /// <param name="commandType">命令类型(存储过程,命令文本或其它.)</param> 
        /// <param name="commandText">T存储过程名称或T-SQL语句</param> 
        /// <param name="commandParameters">SqlParamter参数数组</param> 
        /// <returns>返回影响的行数</returns> 
        public static int ExecuteNonQuery(SqlConnection connection, List<DBCommand> dcs)
        {

            int retval = -1;
            StringBuilder allCommandText = new StringBuilder();
            try
            {
                if (!dcs.HasItem())
                {
                    return retval;
                }
                foreach (var item in dcs)
                {
                    if (item.DbTable.SqlStr.IsNullOrWhiteSpace2())
                    {
                        continue;
                    }
                    allCommandText.AppendLine(item.DbTable.SqlStr);
                }
                if (allCommandText.Length <= 0)
                {
                    return retval;
                }
                //记录db log
                //LogHelper.WriteLog("DBLogSql", allCommandText.ToString());
                if (connection == null) throw new ArgumentNullException("connection");
                if (connection.State != ConnectionState.Open)
                {
                    connection.Open();
                }
                using (connection)
                {
                    // 创建SqlCommand命令,并进行预处理 
                    SqlCommand cmd = new SqlCommand();
                    cmd.Connection = connection;
                    cmd.CommandText = allCommandText.ToString();
                    cmd.CommandTimeout = _CommandTimeout;

                    // Finally, execute the command 
                    retval = cmd.ExecuteNonQuery();
                    connection.Close();
                }
                return retval;
            }
            catch (Exception ex)
            {
                //SystemEvent.GetInstance().OnSystemMessageEvent("数据库访问错误！ExecuteNonQuery" + allCommandText + "---" + ex.ToString(), true, 2);
                throw ex;
            }
        }
        /// <summary>
        /// 处理完成，回写db标记
        /// </summary>
        /// <param name="dcs"></param>
        public static void DBFlagReBack(List<DBCommand> dcs)
        {
            try
            {
                //db处理成功，写dblog
                foreach (var dc in dcs)
                {
                    if (dc != null && dc.Num > -1)
                    {
                        //LogHelper.WriteLog(dc.SuccessDBlogFilePath, $"|{dc.Num}|dbBulk");
                        actionDBPushFlag?.Invoke($"|{dc.Num}|dbBulk");
                    }
                }
            }
            catch (Exception ex)
            {
                //SystemEvent.GetInstance().OnSystemMessageEvent("回写dc.Num DBflag失败，不影响数据库" + ex.ToString(), true, 2);
            }

        }
        #endregion ExecuteNonQuery方法结束
        #region SqlBulkCopy
        /// <summary>
        /// SqlBulkCopy 用于新增
        /// </summary>
        /// <param name="connectionString"></param>
        /// <param name="TableName"></param>
        /// <param name="dt"></param>
        /// <param name="batchSize"></param>
        public static void SqlBulkCopyByDatatableNoColumnMappings(string connectionString, string TableName, DataTable dt, int batchSize = 1000)
        {

            using (SqlBulkCopy sqlbulkcopy = new SqlBulkCopy(connectionString, SqlBulkCopyOptions.Default))
            {
                try
                {
                    _commandQueue.Enqueue(new DBQueueData() { type = 3, logName = String.Intern("DBLogBulkAdd"), commandText = String.Intern(TableName), DataTable = dt.Copy() });

                    sqlbulkcopy.BatchSize = batchSize;
                    sqlbulkcopy.BulkCopyTimeout = _BulkCopyTimeout;
                    sqlbulkcopy.DestinationTableName = TableName;
                    sqlbulkcopy.WriteToServer(dt);
                }
                catch (System.Exception ex)
                {
                    //LogHelper.Error(dt.toJsonStr());
                    throw ex;
                }
            }
        }
        /// <summary>
        /// SqlBulkCopy 用于更新
        /// 表数据，注意去重复（多条相同数据）
        /// </summary>
        /// <param name="conn"></param>
        /// <param name="destinationTableName"></param>
        /// <param name="data"></param>
        /// <param name="columnNameToMatch"></param>
        /// <param name="columnNamesToUpdate"></param>
        /// <param name="batchSize"></param>
        public static void SqlBulkCopyInsertOrUpdate(SqlConnection conn, string destinationTableName, DataTable data,
           string[] columnNameToMatch,
           string[] columnNamesToUpdate, int batchSize = 1000)
        {
            using (conn)
            {
                try
                {
                    //去重
                    GetDistinctSelf(data, columnNameToMatch);

                    bool wasOpen = conn.State == ConnectionState.Open;
                    if (!wasOpen)
                        conn.Open();
                    string tempTablename = "#" + data.TableName + "_" + Guid.NewGuid().ToString("N");
                    CreateTempTable(conn, destinationTableName, tempTablename);

                    _commandQueue.Enqueue(new DBQueueData() { type = 3, logName = String.Intern("DBLogBulkUpdate"), commandText = String.Intern(destinationTableName), DataTable = data.Copy() });

                    using (SqlBulkCopy sbc = new SqlBulkCopy(conn))
                    {
                        sbc.BatchSize = batchSize;
                        sbc.BulkCopyTimeout = _BulkCopyTimeout;
                        sbc.DestinationTableName = tempTablename;
                        sbc.WriteToServer(data);
                    }
                    MergeTempAndDestination(conn, destinationTableName, tempTablename, columnNameToMatch, columnNamesToUpdate);

                    if (!wasOpen)
                        conn.Close();
                }
                catch (System.Exception ex)
                {
                    //LogHelper.Error(data.toJsonStr());
                    throw ex;
                }
            }

        }
        /// <summary>
        /// 去重处理，留下index最大的一条
        /// </summary>
        /// <param name="SourceDt"></param>
        /// <param name="filedName"></param>
        /// <returns></returns>
        public static void GetDistinctSelf(DataTable SourceDt, string[] filedName)
        {
            if (!filedName.HasItem())
            {
                return;
            }
            for (int i = 0; i < SourceDt.Rows.Count; i++)
            {
                var tmpSelectSql = "";
                foreach (var item in filedName)
                {
                    if (tmpSelectSql.IsNullOrWhiteSpace2())
                    {
                        tmpSelectSql = $"{item}='{SourceDt.Rows[i][item]}'";
                        continue;
                    }
                    tmpSelectSql += $" and {item}='{SourceDt.Rows[i][item]}'";
                }
                DataRow[] rows = SourceDt.Select(tmpSelectSql);
                if (rows.Length > 1)
                {
                    SourceDt.Rows.RemoveAt(i);
                }
            }

        }
        /// <summary>
        /// MERGE INTO 更新对应表数据
        /// </summary>
        /// <param name="conn"></param>
        /// <param name="destinationTableName"></param>
        /// <param name="tempTablename"></param>
        /// <param name="matchingColumn">主键</param>
        /// <param name="columnNamesToUpdate"></param>
        private static void MergeTempAndDestination(SqlConnection conn, string destinationTableName, string tempTablename,
           string[] matchingColumns,
           string[] columnNamesToUpdate,
           bool isDropTempTable = true)
        {
            if (conn.State != ConnectionState.Open)
            {
                conn.Open();
            }
            var allColumns = matchingColumns.Concat(columnNamesToUpdate).ToArray();

            var mergeSql = "MERGE INTO " + destinationTableName + " AS Target\r\n" +
                           "USING " + tempTablename + " AS Source\r\n" +
                            BuildJoinConditionsForUpdateOrInsert(matchingColumns, "Source", "Target") + "\r\n" +
                           "WHEN MATCHED THEN\r\n" +
                            BuildUpdateSet(columnNamesToUpdate, "Source", "Target") + "\r\n" +
                           "WHEN NOT MATCHED BY TARGET THEN\r\n" +
                           BuildInsertSet(allColumns, "Source") + ";\r\n";

            if (isDropTempTable && tempTablename.StartsWith("#"))
            {
                mergeSql += "DROP TABLE " + tempTablename + ";\r\n";
            }

            var cmdTempTable = conn.CreateCommand();
            cmdTempTable.CommandText = mergeSql;
            cmdTempTable.ExecuteNonQuery();
        }
        static string BuildJoinConditionsForUpdateOrInsert(string[] updateOn, string sourceAlias = "Source", string targetAlias = "Target")
        {
            StringBuilder command = new StringBuilder();

            command.Append("ON " + "[" + targetAlias + "]" + "." + "[" + updateOn[0] + "]" + " = " + "[" + sourceAlias + "]" + "." + "[" + updateOn[0] + "]" + " ");

            if (updateOn.Length > 1)
            {
                // Start from index 1 to just append "AND" conditions
                for (int i = 1; i < updateOn.Length; i++)
                {
                    command.Append("AND " + "[" + targetAlias + "]" + "." + "[" + updateOn[i] + "]" + " = " + "[" + sourceAlias + "]" + "." + "[" + updateOn[i] + "]" + " ");
                }
            }

            return command.ToString();
        }
        static string BuildUpdateSet(string[] columns, string sourceAlias = "Source", string targetAlias = "Target")
        {
            StringBuilder command = new StringBuilder();
            List<string> paramsSeparated = new List<string>();

            command.Append("UPDATE SET ");

            foreach (var column in columns)
            {
                paramsSeparated.Add("[" + targetAlias + "]" + "." + "[" + column + "]" + " = " + "[" + sourceAlias + "]" + "." + "[" + column + "]");
            }

            command.Append(string.Join(", ", paramsSeparated) + " ");

            return command.ToString();
        }
        static string BuildInsertSet(string[] columns, string sourceAlias = "Source")
        {
            StringBuilder command = new StringBuilder();
            List<string> insertColumns = new List<string>();
            List<string> values = new List<string>();

            command.Append("INSERT (");

            foreach (var column in columns)
            {
                insertColumns.Add("[" + column + "]");
                values.Add("[" + sourceAlias + "]" + "." + "[" + column + "]");
            }

            command.Append(string.Join(", ", insertColumns));
            command.Append(") values (");
            command.Append(string.Join(", ", values));
            command.Append(")");

            return command.ToString();
        }
        private static void CreateTempTable(SqlConnection conn, string destinationTableName, string tempTablename)
        {
            if (conn.State != ConnectionState.Open)
            {
                conn.Open();
            }
            var cmdTempTable = conn.CreateCommand();
            cmdTempTable.CommandText = "SELECT TOP 0 * \r\n" +
                                       "INTO " + tempTablename + "\r\n" +
                                       "FROM " + destinationTableName;
            cmdTempTable.ExecuteNonQuery();
        }
        private static void DropTempTable(SqlConnection conn, string tempTablename)
        {
            if (conn.State != ConnectionState.Open)
            {
                conn.Open();
            }
            var cmdTempTable = conn.CreateCommand();
            cmdTempTable.CommandText = "DROP TABLE " + tempTablename;
            cmdTempTable.ExecuteNonQuery();
        }
        #endregion
        #region ExecuteDataset方法
        /// <summary> 
        /// 执行指定数据库连接对象的命令,返回DataSet. 
        /// </summary> 
        /// <remarks> 
        /// 示例:  
        ///  DataSet ds = ExecuteDataset(conn, CommandType.StoredProcedure, "GetOrders"); 
        /// </remarks> 
        /// <param name="connection">一个有效的数据库连接对象</param> 
        /// <param name="commandType">命令类型 (存储过程,命令文本或其它)</param> 
        /// <param name="commandText">存储过程名或T-SQL语句</param> 
        /// <returns>返回一个包含结果集的DataSet</returns> 
        public static DataSet ExecuteDataset(SqlConnection connection, CommandType commandType, string commandText)
        {
            return ExecuteDataset(connection, commandType, commandText, (SqlParameter[])null);
        }

        /// <summary> 
        /// 执行指定数据库连接对象的命令,指定存储过程参数,返回DataSet. 
        /// </summary> 
        /// <remarks> 
        /// 示例:  
        ///  DataSet ds = ExecuteDataset(conn, CommandType.StoredProcedure, "GetOrders", new SqlParameter("@prodid", 24)); 
        /// </remarks> 
        /// <param name="connection">一个有效的数据库连接对象</param> 
        /// <param name="commandType">命令类型 (存储过程,命令文本或其它)</param> 
        /// <param name="commandText">存储过程名或T-SQL语句</param> 
        /// <param name="commandParameters">SqlParamter参数数组</param> 
        /// <returns>返回一个包含结果集的DataSet</returns> 
        public static DataSet ExecuteDataset(SqlConnection connection, CommandType commandType, string commandText, params SqlParameter[] commandParameters)
        {
            //if (connection == null) throw new ArgumentNullException("connection");
            //Log.Info("调用存储过程：" + commandText);
            //if (commandParameters != null)
            //    foreach (SqlParameter item in commandParameters)
            //    {
            //        if (item.Direction == ParameterDirection.Input)
            //            Log.Info(item.ParameterName + "  " + item.Value.ToString());
            //    }

            try
            {
                // 预处理 
                SqlCommand cmd = new SqlCommand();
                cmd.CommandTimeout = _CommandTimeout;
                bool mustCloseConnection = false;
                PrepareCommand(cmd, connection, (SqlTransaction)null, commandType, commandText, commandParameters, out mustCloseConnection);

                // 创建SqlDataAdapter和DataSet. 
                using (SqlDataAdapter da = new SqlDataAdapter(cmd))
                {
                    DataSet ds = new DataSet();

                    // 填充DataSet. 
                    da.Fill(ds);

                    cmd.Parameters.Clear();

                    if (mustCloseConnection)
                        connection.Close();

                    return ds;
                }
            }
            catch (Exception ex)
            {
                //SystemEvent.GetInstance().OnSystemMessageEvent("数据库访问错误！ExecuteDataset" + ex.ToString(), true, 2);
                DataSet dst = new DataSet();
                DataTable dt = new DataTable();
                dst.Tables.Add(dt);
                return dst;
            }
        }
        #endregion ExecuteDataset数据集命令结束

        #region 持仓表操作
        public static int DelPositionByAccountId(string id)
        {
            System.Data.SqlClient.SqlConnection cnn = new System.Data.SqlClient.SqlConnection(GetConnSting());
            System.Data.SqlClient.SqlCommand cm = new System.Data.SqlClient.SqlCommand();
            cm.Connection = cnn;
            cnn.Open();
            System.Data.SqlClient.SqlTransaction trans = cnn.BeginTransaction();
            try
            {
                cm.CommandText = string.Format("DELETE FROM POSITION WHERE account_no={0}", id);
                cm.ExecuteNonQuery();
                trans.Commit();
                return 0;
            }
            catch
            {
                trans.Rollback();
            }
            finally
            {
                cnn.Close();
                trans.Dispose();
                cnn.Dispose();
            }
            return -1;
        }

        public static int AddPositionByAccountId(DataTable dt)
        {
            System.Data.SqlClient.SqlConnection cnn = new System.Data.SqlClient.SqlConnection(GetConnSting());
            System.Data.SqlClient.SqlCommand cm = new System.Data.SqlClient.SqlCommand();
            cm.Connection = cnn;
            cnn.Open();
            System.Data.SqlClient.SqlTransaction trans = cnn.BeginTransaction();
            try
            {

                foreach (DataRow dr in dt.Rows)
                {
                    cm.CommandText = "update   [表]   set   [数量]   =   @amount   where   productID   =   @productID";
                    cm.Parameters.Add("@amount", SqlDbType.Int);
                    cm.Parameters["@amount"].Value = Convert.ToInt32(dr["amount"]);
                    cm.Parameters.Add("@productID", SqlDbType.VarChar);
                    cm.Parameters["@productID"].Value = dr["productID"].ToString();
                    cm.ExecuteNonQuery();
                }
                trans.Commit();
                return 0;
            }
            catch
            {
                trans.Rollback();
            }
            finally
            {
                cnn.Close();
                trans.Dispose();
                cnn.Dispose();
            }
            return -1;
        }
        #endregion

        public static string FormatSqlParam(string name, params SqlParameter[] commandParameters)
        {
            StringBuilder sb = new StringBuilder();
            sb.Append("[");
            sb.Append(name);
            sb.Append("]");
            sb.AppendLine();
            sb.Append("[");
            try
            {
                for (int i = 0; i < commandParameters.Length; i++)
                {
                    string key = commandParameters[i].ParameterName.ToString();
                    string value = string.Empty;
                    if (commandParameters[i].Value != null)
                    {
                        value = commandParameters[i].Value.ToString();
                    }
                    sb.Append(key + "=" + value).Append(",");
                }
            }
            catch (Exception ex)
            {
                //LogHelper.WriteLog("ErrorLog", "DB log error", ex);
            }
            sb.Append("]");
            return sb.ToString();
        }
        public static string FormatSqlParam(string name, SqlParameterCollection commandParameters)
        {
            StringBuilder sb = new StringBuilder();
            sb.Append("[");
            sb.Append(name);
            sb.Append("]");
            sb.AppendLine();
            sb.Append("[");
            try
            {
                for (int i = 0; i < commandParameters.Count; i++)
                {
                    string key = commandParameters[i].ParameterName.ToString();
                    string value = string.Empty;
                    if (commandParameters[i].Value != null)
                    {
                        value = commandParameters[i].Value.ToString();
                    }
                    sb.Append(key + "=" + value).Append(",");
                }
            }
            catch (Exception ex)
            {
                //LogHelper.WriteLog("ErrorLog", "DB log error", ex);
            }
            sb.Append("]");
            return sb.ToString();
        }
        public static void writeErrorDBLog(SqlParameter para, string proceName)
        {
            if (para.Value.ToNotNullString().IsNotNullOrWhiteSpace())
            {
                //LogHelper.Error(string.Format("[{0}]Error:{1}\r\n", proceName, para.Value.ToNotNullString()));
            }
        }

        #region 队列功能写日志
        private static void stopThread()
        {
            _start = false;
        }
        public static void Dispose()
        {
            stopThread();
        }

        public static void StartThread()
        {
            Thread checkThread = new Thread(workProcess);
            checkThread.IsBackground = true;
            checkThread.Name = "SqlHelperDBlogThread";
            checkThread.Start();
        }
        private static void workProcess()
        {
            //LogHelper.WriteSystemLog("sqldb workProcess start.");
            _start = true;
            while (_start || _commandQueue.Count > 0)
            {
                try
                {

                    DBQueueData command = getCommand();
                    if (command != null)
                    {
                        try
                        {
                            switch (command.type)
                            {
                                case 1:
                                case 2:
                                    //LogHelper.WriteLog(command.logName, FormatSqlParam(command.commandText, command.sqlParameters));
                                    break;
                                case 3:
                                    //LogHelper.WriteLog(command.logName, $"{command.commandText}:{command.DataTable?.Rows?.Count}:{command.DataTable.toJsonStr()}");
                                    break;
                                default:
                                    //LogHelper.WriteLog(command.logName, $"{command.toJsonStr()}");
                                    break;
                            }
                        }
                        catch (Exception ex)
                        {
                            //LogHelper.WriteLog("ErrorLog", "[sqldb Error] " + ex.ToString());
                        }
                    }
                    else
                    {
                        Thread.Sleep(1);
                    }
                }
                catch (System.Exception ex)
                {
                    //LogHelper.Error(ex);
                }

            }

            //LogHelper.WriteSystemLog("sqldb workProcess exit.");
        }
        private static DBQueueData getCommand()
        {
            DBQueueData data = null;
            _commandQueue.TryDequeue(out data);
            return data;
        }
        #endregion
        /// <summary>
        /// 判读数据库队列是否结束
        /// </summary>
        /// <returns></returns>
        public static bool IsDBCommandFinished()
        {
            int tmpCount = _commandQueue.Count;
            if (tmpCount <= 0)
            {
                Thread.Sleep(100);
                tmpCount = _commandQueue.Count;
            }
            return tmpCount > 0;
        }
    }

    //

    //public sealed class SqlParameterHelper
    //{
    //    public static SqlParameter ToSqlPara(string name, RequestPackageEx request)
    //    {
    //        return new SqlParameter("@" + name, request.GetValueByName(name) ?? "0");
    //    }
    //}

    internal class DBQueueData
    {
        /// <summary>
        /// 1：db 2:dbaysc 3:sqlbulk,4:sqlstr
        /// </summary>
        public int type { get; set; }
        public string logName { get; set; }
        public string commandText { get; set; }
        public SqlParameter[] sqlParameters { get; set; }
        public DataTable DataTable { get; set; }
    }
}
